package com.cl.ks.flow.handler;

import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.flow.enums.NodeHandlerEnum;
import com.cl.ks.utils.FieldUtil;
import lombok.extern.java.Log;
import lombok.extern.log4j.Log4j;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;

@Component
@Slf4j
public class SparkNodeHandler extends BaseFlowNodeHandler {


    public SparkNodeHandler() {

    }

    @Override
    public SparkResult process(BaseFlowNodeHandler.ProcessParam processParam) {
        if (!StringUtils.hasLength(processParam.getSparkNode().getNodePath())) {
            log.error("该节点是开始节点或缺少节点路径参数");
            return null;
        }
        SparkResult sparkResult = sparkNodeFactory.getSparkNodeByCode(processParam.getSparkNode().getNodePath()).handle(newSparkParam(processParam));
        return sparkResult;
    }

    public static SparkParam newSparkParam(ProcessParam processParam) {
        SparkParam sparkParam = new SparkParam();
        sparkParam.setNodeExpression(processParam.getSparkNode().getNodeExpression());
        List<SparkResult> sparkResultList = new LinkedList<>(processParam.getSparkNode().getInputSparkResultList().values());

        Comparator<SparkResult> sparkResultComparator = (o1, o2) -> {
            int positionLeft1 = o1.getPositionLeft();
            int positionLeft2 = o2.getPositionLeft();
            return Integer.compare(positionLeft1, positionLeft2);
        };

        sparkResultList.sort(sparkResultComparator);

        sparkParam.setSparkResultList(sparkResultList);
        return sparkParam;
    }


    @Override
    public NodeHandlerEnum getType() {
        return NodeHandlerEnum.SPARK_NODE;
    }
}
